
use anyhow::Result;

use hyper::{Request, Response, Client, body};
use hyper::client::HttpConnector;
use std::collections::HashMap;

use serde::{Deserialize, Serialize};
use std::borrow::Borrow;

use std::ops::Sub;
use sqlx::{Pool, MySql, types};
use chrono;

use crate::miner::{JsonRpcResponse, JsonRpcError};
use crate::miner;
use crate::db;
use crate::miner::types::TaskType;
use crate::miner::worker::SealWorkerStats;
use crate::db::CommonWorkerLimit;

pub mod check;

pub async fn auto_pledge_sector(pool: &Pool<MySql>, client: &Client<HttpConnector>, url: &str, token: &str) -> Result<()> {
    let window_count = 5;
    let mut miner = 1000u64;
    let config: Vec<CommonWorkerLimit> = db::query_worker_task_limit(pool)?;
    let mut worker_config = HashMap::<String, i32>::new();
    for w in config {
        worker_config.insert(w.worker_id, w.task_limit);
    }
    let rsp = miner::worker::query_worker_from_miner(client, url, token).await?;
    if let Some(e) = rsp.error {
        log::info!("receive error from miner when query workers, {:?}", e);
        return Ok(())
    }
    let workers = rsp.result.unwrap();
    if workers.is_empty(){
        log::debug!("no worker running with miner.")
    }
    let rsp = miner::jobs::query_jobs_from_miner(client, &url, &token).await?;
    if let Some(e) = rsp.error {
        log::debug!("receive error from miner when query jobs, {:?}", e);
        return Ok(())
    }
    let all_jobs = rsp.result.unwrap();

    let mut pledge_workers = Vec::<SealWorkerStats>::new();
    for (id, worker) in workers {
        if !worker.enabled {
            log::warn!("worker with id {} is disable.", id);
            continue
        }

        let limit = worker_config.get(&id);
        if limit.is_none() {
            log::warn!("worker with id {} not support new PC1.", id);
            continue
        }

        let worker_info = db::query_worker_info_with_id(pool, id.as_str()).await?;
        if worker_info.is_none() {
            log::warn!("worker with id {} not found in db.", id);
            continue
        }
        let worker_info = worker_info.unwrap();
        if worker_info.accept_new != 1 {
            log::info!("worker with id {} can not accept new task.", id);
            continue
        }

        let jobs = all_jobs.get(&id);
        if jobs.is_none(){
            pledge_workers.push(worker);
            continue
        }
        let jobs = jobs.unwrap();

        if jobs.is_empty(){
            pledge_workers.push(worker);
            continue
        }

        let limit = limit.unwrap();
        let mut running = 0;
        let mut assigned = 0;
        let mut window = 0;
        for j in jobs {
            if miner == 0 {
                miner = j.sector.miner
            }
            let f = match j.task {
                TaskType::AddPiece => true,
                TaskType::PreCommit1 => true,
                _ => false,
            };
            if f {
                if j.run_wait > 0 {
                    assigned = assigned + 1;
                    if j.run_wait > window {
                        window = j.run_wait
                    }
                }
                if j.run_wait == 0 {
                    running = running + 1;
                }
            }
        }
        // println!("worker {}, limit {}, running {}, assigned {}, window {}", id, limit, running, assigned, window);
        if assigned > 0 {
            log::debug!("worker with id {} have assigned task {}", id, assigned);
            continue
        }
        if window >= window_count {
            log::debug!("worker with id {} have no more window.", id);
            continue
        }
        if running >= *limit {
            log::debug!("worker with id {} have enough task running {}, limit {}", id, running, limit);
            continue
        }
        let count = db::query_worker_running_sector_count(pool, &worker.info.hostname).await?;
        if count as u32 >= *limit {
            log::debug!("worker with id {} have {} running record in db.", id, count);
            continue
        }
        //can pledge one sector
        pledge_workers.push(worker);
    }

    if pledge_workers.is_empty(){
        log::info!("no worker can handle new more task.")
    } else {
        log::info!("pledge length {}", pledge_workers.len());
    }

    // let mut packing_sector = vec![];
    // let packing = miner::sector::query_sector_list(client, url, token, Some(vec!["Packing"])).await?;
    //
    // if let Some(e) = packing.error{
    //     log::debug!("query packing sectors fail. {:?}", e);
    // } else {
    //     if let Some(vec) = packing.result {
    //         packing_sector = vec
    //     }
    // }
    //
    // let mut index = 0;
    // let mut assign_worker = Vec::<String>::new();
    // log::info!("packing sector length {}", packing_sector.len());
    // for sector in packing_sector {
    //     let worker = db::is_sector_assigned_to_worker(pool, sector, miner).await?;
    //     println!("sector {}, worker {:?}", sector, worker);
    //     if worker.is_none(){
    //         if index < pledge_workers.len(){
    //             let id = pledge_workers[index].clone();
    //             let row = db::pledge_sector_for_worker(pool, &id, sector, miner).await?;
    //             log::info!("pledge sector {} to worker {} success. row {}", sector, &id, row);
    //             index = index + 1;
    //         }
    //     }
    // }

    let length = pledge_workers.len();

    for worker in pledge_workers {
        let rsp = miner::pledge::pledge_on_sector(client, url, token).await?;
        if let Some(e) = rsp.error {
            log::error!("receive error from miner when pledge sector, {:?}", e);
        } else {
            let sector = rsp.result.unwrap();

            // let query = "select * from sector where sector_number = ?";
            // let info = sqlx::query_as::<_, db::Sector>(query).bind(sector.number).fetch_all(pool).await?;
            // println!("{:?}", info);
            // log::info!("pledge sector {:?}", sector);
            let row = db::pledge_sector_for_worker(pool, &worker.info.hostname, sector.number, sector.miner).await?;
            log::info!("pledge sector {} to worker {} success. row {}", sector.number, &worker.info.hostname, row);
        }
    }

    log::info!("{} sector pledge", length);
    Ok(())
}